package com.xzx.flink.tableapi;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * 自定义UDF表函数
 *
 * @version 1.0
 * @auther xinzhixuan
 * @date 2022/6/11 21:58
 */
public class TableAPI_06_UDF_TableFunction {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> stream = env.fromElements("hello flink", "hello java", "hello sql");
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.createTemporaryView("word_table", stream, Schema.newBuilder()
                .columnByExpression("line", $("f0"))
                .build());

        tableEnv.createTemporarySystemFunction("split", SplitFunction.class);

        tableEnv.toDataStream(tableEnv.sqlQuery("select line, word, len from word_table, lateral table(split(line))")).print("-->");
        tableEnv.toDataStream(tableEnv.sqlQuery("select line, word, len from word_table left join lateral table(split(line)) on true")).print("==>");
        tableEnv.toDataStream(tableEnv.sqlQuery("select line, new_word, new_len from word_table left join lateral table(split(line)) as t(new_word, new_len) on true")).print("**>");

        env.execute();

    }

    @FunctionHint(output = @DataTypeHint("ROW<word STRING, len INT>"))
    public static class SplitFunction extends TableFunction<Row> {

        // 底层架构约定好的名字
        public void eval(String str) {
            for (String s : str.split(" ")) {
                // 使用 父类的collect()方法发送一行数据,返回的这行数据有2个字段
                collect(Row.of(s, s.length()));
            }
        }

    }
}


